package com.sdses.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkSQL_MySQLCDC2Doris_View {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id INT,\n" +
                "    mode STRING,\n" +
                "    voltage FLOAT,\n" +
                "    resistance FLOAT,\n" +
                "    temperature FLOAT,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = '192.168.102.155',\n" +
                "    'port' = '33066',\n" +
                "    'username' = 'datacube_test',\n" +
                "    'password' = 'Gbd#XERA@19d1ZtD',\n" +
//                "    'server-time-zone' = '+08:00',\n" +
                "    'server-time-zone' = 'Asia/Shanghai',\n" +
                "    'database-name' = 'db_datacube_test',\n" +
                "    'table-name' = 'water_sensor_view'\n" +
//                "    'scan.startup.mode' = 'latest-offset',\n" +
//                "    'scan.startup.mode' = 'timestamp',\n" +
//                "    'scan.startup.timestamp-millis' = '1689256818000',\n" +
//                "    'server-id' = '8637102'\n" +
                "  )");

        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
                "LIKE sourceTable (EXCLUDING ALL)");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH (\n" +
//                "'connector' = 'doris',\n" +
//                "'fenodes' = '192.168.102.114:38030',\n" +
//                "'table.identifier' = 'db_jw_test.water_sensor_test',\n" +
//                "'username' = 'datacube_doris_test',\n" +
//                "'password' = 'Nx8Tz_Y8v3x',\n" +
//                "'sink.properties.format' = 'json',\n" +
//                "'sink.properties.read_json_by_line' = 'true',\n" +
//                "'sink.enable-delete' = 'true',\n" +
//                "'sink.label-prefix' = 'doris_label123'\n" +
//                ") LIKE sourceTable (EXCLUDING ALL)");


        tableEnv.executeSql("insert into sinkTable select * from sourceTable ");


    }
}
